package com.pingsky.apm.agent.core.service.sender;

import com.pingsky.apm.agent.api.config.AgentConfig;
import com.pingsky.apm.agent.api.log.PLogger;
import com.pingsky.apm.agent.api.log.PLoggerFactory;
import com.pingsky.apm.agent.core.service.ServiceManager;
import com.pingsky.apm.agent.core.service.message.MessageSerializer;
import com.pingsky.apm.agent.core.service.message.TopicMessage;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.BytesSerializer;
import org.apache.kafka.common.serialization.StringSerializer;
import org.apache.kafka.common.utils.Bytes;

import java.util.Properties;

public class KafkaDataSender implements DataSender {

    private Producer<String, Bytes> producer;
    private MessageSerializer messageSerializer;

    private PLogger logger = PLoggerFactory.getLogger(getClass());

    @Override
    public void send(Object message) {
        TopicMessage topicMessage = messageSerializer.toMessage(message);
        producer.send(new ProducerRecord<String, Bytes>(topicMessage.getTopic(), Bytes.wrap(topicMessage.getMessage().toByteArray())));
    }

    @Override
    public void prepare() throws Throwable {
        AgentConfig agentConfig = ServiceManager.INSTANCE.findService(AgentConfig.class);
        this.messageSerializer = ServiceManager.INSTANCE.findService(MessageSerializer.class);
        try {
            Properties props = new Properties();
            props.put("bootstrap.servers", agentConfig.readString("kafka.servers", ""));
            final Thread currentThread = Thread.currentThread();
            final ClassLoader before = currentThread.getContextClassLoader();
            currentThread.setContextClassLoader(KafkaDataSender.class.getClassLoader());
            try {
                this.producer = new KafkaProducer<String, Bytes>(props, new StringSerializer(), new BytesSerializer());
            } finally {
                currentThread.setContextClassLoader(before);
            }
        } catch (Exception e) {
            logger.error("[数据发送] KAFKA初始化失败", e);
        }
    }

    @Override
    public void start() throws Throwable {

    }
}
